package server;

import common.BlockInfo;
import common.Utils;
import hdfs.hdfsPut;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import interfaces.DataNodeDef;
import interfaces.NameNodeDef;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import transport.WriteBlockTransPorter;

import static common.Utils.*;
import static common.StateCode.*;

public class DataNode extends UnicastRemoteObject implements DataNodeDef {

    private static final Logger logger = LoggerFactory.getLogger(DataNode.class);
    public String blockStorage;
    public String blockInfoStorage;
    public Vector<Integer> blockInfos = new Vector<>();
    public Map<Integer, Integer> blockDetails = new HashMap<>();
    private String IP, NN_IP;
    private int PORT,NN_PORT;
    private int reportDelay;
    private int reportPeriod;
    private final int T_PORT;
    private final WriteBlockTransPorter writeBlockTransPorter;

    public DataNode(String ip,int port,int T_PORT,String storage) throws RemoteException {
        super();
        this.IP = ip;
        this.PORT = port;
        this.NN_IP = Utils.NN_IP;
        this.NN_PORT = Utils.NN_PORT;
        this.T_PORT = T_PORT;
        blockStorage = storage;
        blockInfoStorage = storage+"blockInfo";
        init();
        writeBlockTransPorter = new WriteBlockTransPorter(T_PORT, storage);
        writeBlockTransPorter.startServer();
    }

    public void report(){
        reportDelay = 1000*60*10;
        reportPeriod = 1000*60*60;
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                sendBlockReport(new byte[0]);
                logger.info(IP+" "+PORT+"report");
            }
        },reportDelay,reportPeriod);
    }

    public void init(){
        try {
            //创建存储目录
            File folder = new File(blockStorage);
            if(!folder.exists()){
                Path path = Paths.get(blockStorage);
                Files.createDirectory(path);
                logger.info("创建存储路径");
            }
            //TODO nameNode里block是不可以用的，需要删掉
            //1.把文件中的信息全部读取到内存中
            File file = new File(blockInfoStorage);
            if(file.exists()){
                FileReader fr = new FileReader(file);
                BufferedReader reader = new BufferedReader(fr);
                String str;
                blockInfos=new Vector<>();
                blockDetails=new HashMap<>();
                while((str=reader.readLine())!=null){
                    String[] split = str.split(",");
                    Integer blockNum = Integer.parseInt(split[0]);
                    Integer size = Integer.parseInt(split[1]);
                    blockInfos.add(blockNum);
                    blockDetails.put(blockNum,size);
                }
            }
            //开始发送汇报
            report();
            //开始发送心跳
            heartBeat(new byte[0]);
        } catch (IOException e) {
            logger.info("读取blockDetails失败");
            e.printStackTrace();
        }
    }

    @Override
    public byte[] sendBlockReport(byte[] inp) {
        //检查文件
        Set<Integer> set = checkAllBlock();
        if(set.size()==0){
            logger.info("datanode"+IP+":"+PORT+"所有block未发生篡改");
        }
        hdfsPut.BlockReportRequest.Builder blockReportRequest = hdfsPut.BlockReportRequest.newBuilder();
        blockReportRequest.setIP(IP);
        blockReportRequest.setPORT(PORT);
        for(Integer i:blockInfos){
            hdfsPut.BlockInfo.Builder blockInfo = hdfsPut.BlockInfo.newBuilder();
            blockInfo.setBlockNum(i);
            if(set.contains(i)){   //该block发生篡改
                //TODO 状态码
                blockInfo.setStatus(101);
            }else blockInfo.setStatus(1);
            blockReportRequest.addBlocks(blockInfo);
        }
        NameNodeDef NNStub = getNNStub(NN_IP, NN_PORT);
        try {
            assert NNStub != null;
            byte[] bytes = NNStub.blockReport(blockReportRequest.build().toByteArray());
            try {
                hdfsPut.BlockReportResponse blockReportResponse = hdfsPut.BlockReportResponse.parseFrom(bytes);
                for (hdfsPut.ReadBlockRequest readBlockRequest : blockReportResponse.getReadBlockRequestList()) {
                    DataNodeDef dnStub = getDNStub(readBlockRequest.getIP(), readBlockRequest.getPORT());
                    assert dnStub != null;
                    hdfsPut.ReadBlockResponse readBlockResponse =
                            hdfsPut.ReadBlockResponse.parseFrom(dnStub.readBlock(readBlockRequest.toByteArray()));
                    int blockNum = readBlockRequest.getBlockNo();
                    int size = blockDetails.get(blockNum);
                    writeBlockLocal(readBlockResponse.getData().toByteArray(),blockNum,size);
                }
            } catch (InvalidProtocolBufferException e) {
                e.printStackTrace();
            }
        } catch (RemoteException e) {
            e.printStackTrace();
        }
        return new byte[0];
    }

    @Override
    public byte[] readBlock(byte[] inp) throws RemoteException {
        try {
            hdfsPut.ReadBlockRequest readBlockRequest = hdfsPut.ReadBlockRequest.parseFrom(inp);
            int blockNum = readBlockRequest.getBlockNo();
            String blockPath = blockStorage + blockNum;
            try {
                BufferedInputStream stream = new BufferedInputStream(new FileInputStream(blockPath));
                hdfsPut.ReadBlockResponse.Builder readBlockResponse = hdfsPut.ReadBlockResponse.newBuilder();
                int size = blockDetails.get(blockNum);
                byte[] bytes = new byte[size];
                try {
                    stream.read(bytes);
                    stream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                readBlockResponse.setData(ByteString.copyFrom(bytes));
                readBlockResponse.setStatus(fileOk);
                return readBlockResponse.build().toByteArray();
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        return new byte[0];
    }

    public byte[] writeBlock_bck(byte[] inp) throws RemoteException {
        try {
            hdfsPut.WriteBlockRequest writeReq = hdfsPut.WriteBlockRequest.parseFrom(inp);
            hdfsPut.BlockLocations blockInfo = writeReq.getBlockInfo();
            // 取出数据
            int blockNumber = blockInfo.getBlockNumber();
            ByteString data = writeReq.getData();
            int replicate = writeReq.getReplicate();
            int validLen = writeReq.getSize();
            blockDetails.put(blockNumber,validLen);
            // 写到本地
            writeBlockLocal(data.toByteArray(),blockNumber,validLen+4);
            if(replicate>1){
                // 发给下游
                hdfsPut.WriteBlockRequest.Builder request = hdfsPut.WriteBlockRequest.newBuilder();
                request.setReplicate(replicate-1);
                request.setBlockInfo(blockInfo);
                request.setData(data);
                request.setSize(validLen);
                hdfsPut.DataNodeLocation locations = blockInfo.getLocations(REPLICATE_NUM - replicate + 1);
                System.out.printf("%d -> %d\n", PORT,locations.getPort());
                DataNodeDef dnStub = getDNStub(locations.getIp(), locations.getPort());
                assert dnStub != null;
                return dnStub.writeBlock(request.build().toByteArray());
            }else if(replicate==1){
                // 自己是最后一个
                if(checkSum(data.toByteArray())){
                    hdfsPut.WriteBlockResponse.Builder response = hdfsPut.WriteBlockResponse.newBuilder();
                    response.setStatus(blockOk);
                    return response.build().toByteArray();
                }
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        hdfsPut.WriteBlockResponse.Builder response = hdfsPut.WriteBlockResponse.newBuilder();
        response.setStatus(blockError);
        return response.build().toByteArray();
    }

    @Override
    public byte[] writeBlock(byte[] inp) throws RemoteException {

        try {
            hdfsPut.WriteBlockRequest writeReq = hdfsPut.WriteBlockRequest.parseFrom(inp);
            hdfsPut.BlockLocations blockInfo = writeReq.getBlockInfo();
            // 取出数据
            int blockNumber = blockInfo.getBlockNumber();
            blockInfos.add(blockNumber);
            logger.info("DATANODE"+IP+":"+PORT+"写块"+blockNumber);
            ByteString data = writeReq.getData();
            int replicate = writeReq.getReplicate();
            // 写到本地
            writeBlockLocal(data.toByteArray(),blockNumber,writeReq.getSize());
            if(replicate>1){
                // 发给下游
                hdfsPut.WriteBlockRequest.Builder request = hdfsPut.WriteBlockRequest.newBuilder();
                request.setReplicate(replicate-1);
                request.setBlockInfo(blockInfo);
                request.setData(data);
                request.setSize(writeReq.getSize());
                hdfsPut.DataNodeLocation locations = blockInfo.getLocations(REPLICATE_NUM - replicate + 1);
//                System.out.printf("%d->%d\n", PORT,locations.getPort());
                DataNodeDef dnStub = getDNStub(locations.getIp(), locations.getPort());
                assert dnStub != null;
                return dnStub.writeBlock(request.build().toByteArray());
            }else if(replicate==1){
                // 自己是最后一个
                if(checkSum(data.toByteArray())){
                    hdfsPut.WriteBlockResponse.Builder response = hdfsPut.WriteBlockResponse.newBuilder();
                    response.setStatus(blockOk);
                    return response.build().toByteArray();
                }
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        hdfsPut.WriteBlockResponse.Builder response = hdfsPut.WriteBlockResponse.newBuilder();
        response.setStatus(blockError);
        return response.build().toByteArray();
    }

    @Override
    public byte[] heartBeat(byte[] inp) throws RemoteException {
        try {
            hdfsPut.heartBeatRequest heartBeatRequest = hdfsPut.heartBeatRequest.parseFrom(inp);
            byte[] bytes = hdfsPut.heartBeatResponse.newBuilder().setPONG("PONG").build().toByteArray();
            if(heartBeatRequest.getPING().equals("PING")){
                return bytes;
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        return new byte[0];
    }

    @Override
    public byte[] transPort(byte[] inp) throws RemoteException{
        try {
            hdfsPut.TransPortRequest req = hdfsPut.TransPortRequest.parseFrom(inp);
            hdfsPut.BlockLocations blockInfo = req.getBlockInfo();
            // 取出数据
            int blockNumber = blockInfo.getBlockNumber();
            int replicate = req.getReplicate();
            int validLen = req.getSize();
            int idx = REPLICATE_NUM-replicate;
            blockDetails.put(blockNumber,validLen);
            if(replicate>1){
                // 发给下游
                hdfsPut.DataNodeLocation nextDn = blockInfo.getLocations(idx + 1);
                hdfsPut.TransPortRequest.Builder request = hdfsPut.TransPortRequest.newBuilder();
                request.setReplicate(replicate-1);
                request.setBlockInfo(blockInfo);
                request.setSize(validLen);
                logger.info("{}->{}", PORT,nextDn.getPort());
                DataNodeDef dnStub = getDNStub(nextDn.getIp(), nextDn.getPort());
                assert dnStub != null;
                logger.info("transPorting");
                byte[] out = dnStub.transPort(request.build().toByteArray());
                logger.info("transPort finished");
                // 连接下游
                writeBlockTransPorter.connect("localhost",nextDn.getTPort());
                // TODO 考虑连接未建立情况
                // 获取下游信息
                hdfsPut.TransPortResponse response = hdfsPut.TransPortResponse.parseFrom(out);
                if(response.getStatus()==connectOk){
                    return out;
                }else{
                    return hdfsPut.TransPortResponse.newBuilder().setStatus(connectError).build().toByteArray();
                }

            }else if(replicate==1){
                hdfsPut.TransPortResponse.Builder rep = hdfsPut.TransPortResponse.newBuilder();
                writeBlockTransPorter.addBridge();
                return rep.setStatus(connectOk).build().toByteArray();
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
            hdfsPut.WriteBlockResponse.Builder response = hdfsPut.WriteBlockResponse.newBuilder();
            response.setStatus(connectError);
            return response.build().toByteArray();
        }
        hdfsPut.WriteBlockResponse.Builder response = hdfsPut.WriteBlockResponse.newBuilder();
        response.setStatus(connectError);
        return response.build().toByteArray();
    }

    /**
     *  将块写入文件 只写入size+4大小
     * @param data 字节数组
     * @param blockNum 块序号
     * @throws IOException IO异常
     */

    private boolean writeBlockLocal(byte[] data,long blockNum,int size){
        try {
            FileOutputStream foutStream = new FileOutputStream(blockStorage + blockNum);
            BufferedOutputStream boutStream = new BufferedOutputStream(foutStream);
            boolean isSyn = false;
            if(data.length==size){
                data = unitBytes(data,addCRC(data));
                isSyn=true;
            }
            boutStream.write(data,0,size+4);
            blockDetails.put((int)blockNum,size);
            logger.info("写入block大小"+data.length);
            logger.info(IP+PORT+"save block"+blockNum);
            boutStream.close();
            //写入block状态,如果是同步写则不需要该过程
            if(!isSyn){
                FileWriter stream = new FileWriter(blockInfoStorage,true);
                String str = blockNum+","+size+"\n";
                stream.append(str);
                stream.close();
            }
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    private Set<Integer> checkAllBlock(){
        logger.info("datanode"+IP+":"+PORT+"对所有block进行检查");
        Set<Integer> map = new HashSet<>();
        for (Integer blockNum : blockInfos) {
            if(!checkBlock(blockNum)){
                map.add(blockNum);
            }
        }
        if(map.size()==0){
            logger.info("datanode"+IP+":"+PORT+"全部block正常");
        }else{
            logger.info("datanode"+IP+":"+PORT+"部分block被篡改，开始进行同步");
        }
        return map;
    }

    private boolean checkBlock(int blockNum){
        logger.info("datanode"+IP+PORT+"对block"+blockNum+"进行检查"+"文件名"+blockStorage+blockNum);
        FileInputStream fileInputStream;
        try {
            fileInputStream = new FileInputStream(blockStorage + blockNum);
            BufferedInputStream inputStream = new BufferedInputStream(fileInputStream);
            byte[] buffer = new byte[blockDetails.get(blockNum)+4];
            inputStream.read(buffer);
            inputStream.close();
            boolean checkSum = checkSum(buffer);
            if(!checkSum){
                logger.info("datanode"+IP+":"+PORT+"block"+blockNum+"发生篡改");
                File file = new File(blockStorage+blockNum);
                if(file.delete()){
                    logger.info("datanode"+IP+":"+PORT+"block"+blockNum+"已经被删除");
                }else{
                    logger.info("datanode"+IP+":"+PORT+"block"+blockNum+"未被删除");
                }
            }else{
                logger.info("datanode"+IP+":"+PORT+"block"+blockNum+"未发生篡改");
            }
            return checkSum;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }
}
